#include "zhelpers.h"
#include "proto.msg.pb.h"
#include "Log.h"

#include "front_thread.h"
#include "front_server.h"
#include "vs_opencv.h"

int front_thread::thread_func()
{
	if(get_thread_pool() == NULL){
		LOG_ERROR_VA("[%s] pool_work_thread don't belong to a thread_pool", get_name());
		return -1;
	}
	int ret;
	void *receiver = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REP);
	if(receiver == NULL){
		LOG_ERROR_VA("[%s] zmq_socket failed", get_name());
		return -1;
	}
	ret = zmq_connect(receiver, "tcp://199.155.122.11:45510"); //front back_address
	if(ret == -1){
		LOG_ERROR_VA("[%s] front back address connect failed", get_name());
		return -1;
	}
	
	//gf_extract
	void *gf_extract_zmq = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REQ);
	if(gf_extract_zmq == NULL){
		LOG_ERROR_VA("[%s] gf_extract_zmq zmq_socket failed", get_name());
		return -1;
	}
	string gf_extract_tcp_addr = "tcp://";
	gf_extract_tcp_addr.append(get_gf_extract().get_ip());
	gf_extract_tcp_addr.append(":");
	gf_extract_tcp_addr.append(get_gf_extract().get_port());
	ret = zmq_connect(gf_extract_zmq, gf_extract_tcp_addr.c_str());
	if(ret == -1){
		LOG_ERROR_VA("[%s] gf_extract_zmq connect failed", get_name());
		return -1;
	}

	//gf_merge(include all match)
	void *gf_merge_zmq = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REQ);
	if(gf_merge_zmq == NULL){
		LOG_ERROR_VA("[%s] gf_merge_zmq zmq_socket failed", get_name());
		return -1;
	}
	string gf_merge_tcp_addr = "tcp://";
	gf_merge_tcp_addr.append(get_gf_merge().get_ip());
	gf_merge_tcp_addr.append(":");
	gf_merge_tcp_addr.append(get_gf_merge().get_port());
	ret = zmq_connect(gf_merge_zmq, gf_merge_tcp_addr.c_str());
	if(ret == -1){
		LOG_ERROR_VA("[%s] gf_merge_zmq connect failed", get_name());
		return -1;
	}

	LOG_DEBUG_VA("[%s] thread start", get_name());
	while(!(get_thread_pool()->has_stoped())){
		//Wait for request from client
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("[%s] thread get task...", get_name());
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::front_req front_req;
		if(!front_req.ParseFromString(proto_buf)){
			LOG_ERROR_VA("ParseFromString fail");
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] recv length: %d, class_id: %d", get_name(), size, front_req.class_id());
		//the work of engine
		char *path_buf = (char *)malloc(front_req.image_data().size());
		memcpy(path_buf, front_req.image_data().c_str(), front_req.image_data().size());
		IplImage *ipl = cv_load_image(path_buf, front_req.image_data().size());
		delete [] path_buf;
		if(ipl == NULL){
			LOG_ERROR_VA("[%s] open_cv load error" ,get_name());
			send_error_reply(receiver);
			continue;
		}
	
		//send message to gf_extract
		proto::extract_req extract_req;
		extract_req.mutable_head()->set_version(420);
		extract_req.mutable_head()->set_command(420);
		extract_req.mutable_head()->set_password("xxxxxxxx", 8);
		extract_req.mutable_head()->set_type(0);
		extract_req.mutable_head()->set_status(0);
		extract_req.mutable_head()->set_src(1);
		extract_req.mutable_head()->set_dest(2);
		extract_req.mutable_head()->set_sn_high(123456789);
		extract_req.mutable_head()->set_sn_low(123456789);
		extract_req.mutable_head()->set_length(99);

		extract_req.set_class_id(front_req.class_id());

		LOG_DEBUG_VA("image width: %d, height: %d, depth: %d, channel: %d, size: %d", ipl->width,ipl->height, ipl->depth,ipl->nChannels,ipl->imageSize);

		extract_req.set_image_width(ipl->width);
		extract_req.set_image_height(ipl->height);
		extract_req.set_image_depth(ipl->depth);
		extract_req.set_image_channel(ipl->nChannels);
		extract_req.set_image_size(ipl->imageSize);
		extract_req.set_image_data(ipl->imageData, ipl->imageSize);
		extract_req.set_text_count(0);

		proto_buf.clear();
		if(!extract_req.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}
		//request gf_extract 
		zmq_msg_t extract_req_msg;
		zmq_msg_init_size(&extract_req_msg, proto_buf.size());
		memcpy(zmq_msg_data(&extract_req_msg), proto_buf.c_str(), proto_buf.size());
		LOG_DEBUG_VA("extract req size: %d", proto_buf.size());
		size = zmq_sendmsg(gf_extract_zmq, &extract_req_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_sendmsg to gf error");
			send_error_reply(receiver);
			continue;
		}
		zmq_msg_t gf_extract_resp_msg;
		zmq_msg_init(&gf_extract_resp_msg);
		size = zmq_recvmsg(gf_extract_zmq, &gf_extract_resp_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		proto_buf.clear();
		proto_buf.assign((char *)zmq_msg_data(&gf_extract_resp_msg), size);
		proto::gf_extract_resp gf_extract_resp;
		if(!gf_extract_resp.ParseFromString(proto_buf)){
			LOG_ERROR_VA("ParseFromString fail");
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] recv length: %d", get_name(), size);
		//request gf_merge (gf_match)
		proto::gf_match_req gf_match_req;
		gf_match_req.mutable_head()->set_version(420);
		gf_match_req.mutable_head()->set_command(420);
		gf_match_req.mutable_head()->set_password("xxxxxxxx", 8);
		gf_match_req.mutable_head()->set_type(0);
		gf_match_req.mutable_head()->set_status(0);
		gf_match_req.mutable_head()->set_src(1);
		gf_match_req.mutable_head()->set_dest(2);
		gf_match_req.mutable_head()->set_sn_high(123456789);
		gf_match_req.mutable_head()->set_sn_low(123456789);
		gf_match_req.mutable_head()->set_length(99);
	
		gf_match_req.set_class_id(front_req.class_id());
		gf_match_req.set_b2c_flag(front_req.b2c_flag());
		gf_match_req.set_res_num(front_req.gf_num());
		gf_match_req.set_merge_same(front_req.merge_same());
		gf_match_req.set_vid_num(gf_extract_resp.vids_size());
		for(int i = 0; i < gf_extract_resp.vids_size(); i++)
			gf_match_req.add_vids(gf_extract_resp.vids(i));
		gf_match_req.set_feature_data(gf_extract_resp.feature_data());
		proto_buf.clear();
		if(!gf_match_req.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}
		zmq_msg_t gf_match_req_msg;
		zmq_msg_init_size(&gf_match_req_msg, proto_buf.size());
		memcpy(zmq_msg_data(&gf_match_req_msg), proto_buf.c_str(), proto_buf.size());
		size = zmq_sendmsg(gf_merge_zmq, &gf_match_req_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_sendmsg to gf error");
			send_error_reply(receiver);
			continue;
		}
		
		zmq_msg_t gf_merge_resp_msg;
		zmq_msg_init(&gf_merge_resp_msg);
		size = zmq_recvmsg(gf_merge_zmq, &gf_merge_resp_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		proto_buf.clear();
		proto_buf.assign((char *)zmq_msg_data(&gf_merge_resp_msg), size);
		proto::gf_merge_resp gf_merge_resp;
		if(!gf_merge_resp.ParseFromString(proto_buf)){
			LOG_ERROR_VA("ParseFromString fail");
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] recv length: %d", get_name(), size);

		//send message to client
		proto::front_resp front_resp;
		front_resp.mutable_head()->set_version(420);
		front_resp.mutable_head()->set_command(420);
		front_resp.mutable_head()->set_password("xxxxxxxx", 8);
		front_resp.mutable_head()->set_type(1);
		front_resp.mutable_head()->set_status(0);
		front_resp.mutable_head()->set_src(1);
		front_resp.mutable_head()->set_dest(2);
		front_resp.mutable_head()->set_sn_high(123456789);
		front_resp.mutable_head()->set_sn_low(123456789);
		front_resp.mutable_head()->set_length(99);

		front_resp.set_res_num(gf_merge_resp.res_num());
		for(int i = 0; i < gf_merge_resp.res_num(); i++){
			front_resp.add_iid(gf_merge_resp.res(i).imageid());
			front_resp.add_gid(gf_merge_resp.res(i).groupid());
			front_resp.add_simi(gf_merge_resp.res(i).similarity());
			front_resp.add_type(gf_merge_resp.res(i).imagetype());
			front_resp.add_source(gf_merge_resp.res(i).source());
		}

		proto_buf.clear();
		if(!front_resp.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}
		zmq_msg_t front_resp_msg;
		zmq_msg_init_size(&front_resp_msg, proto_buf.size());
		memcpy(zmq_msg_data(&front_resp_msg), proto_buf.c_str(), proto_buf.size());
		//send message to client
		size = zmq_sendmsg(receiver, &front_resp_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_sendmsg to client error");
			send_error_reply(receiver);
			continue;
		}

		for(int i = 0; i < 5; i++){
			LOG_DEBUG_VA("[%s] result iid: %llu, gid: %llu, simi: %d, source: %d", get_name(),
					gf_merge_resp.res(i).imageid(), gf_merge_resp.res(i).groupid(), gf_merge_resp.res(i).similarity(), gf_merge_resp.res(i).source());
		}
	}
	LOG_INFO_VA("[%s] stoped",get_name());
	zmq_close(receiver);
	return 0;
}

address front_thread::get_gf_extract(){
	front_server *server = dynamic_cast<front_server *>(get_thread_pool()->get_server());
	return server->get_gf_extract();
	//return get_thread_pool()->get_server()->get_one_gf_server();
}

address front_thread::get_gf_merge(){
	front_server *server = dynamic_cast<front_server *>(get_thread_pool()->get_server());
	return server->get_gf_merge();
	//return get_thread_pool()->get_server()->get_one_gf_server();
}

address front_thread::get_lf_extract(){
	front_server *server = dynamic_cast<front_server *>(get_thread_pool()->get_server());
	return server->get_lf_extract();
}

address front_thread::get_lf_merge(){
	front_server *server = dynamic_cast<front_server *>(get_thread_pool()->get_server());
	return server->get_lf_merge();
}


